package com.sprucetec.live.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.sprucetec.live.annotation.GuavaTateLimeter;
import com.sprucetec.live.annotation.LiveRedisCache;
import com.sprucetec.live.cache.JvmCache;
import com.sprucetec.live.common.dto.PageDTO;
import com.sprucetec.live.common.dto.PagedListDTO;
import com.sprucetec.live.common.dto.business.LiveUserDto;
import com.sprucetec.live.common.dto.business.LiveVideoDto;
import com.sprucetec.live.common.dto.business.NettyMessageDto;
import com.sprucetec.live.common.exceptioin.SprucetecException;
import com.sprucetec.live.common.webcontext.LoginUserContext;
import com.sprucetec.live.constants.MqConstant;
import com.sprucetec.live.constants.RedisKeyConstant;
import com.sprucetec.live.constants.SprucetecConstant;
import com.sprucetec.live.entity.*;
import com.sprucetec.live.enums.*;
import com.sprucetec.live.mapper.LiveVideoMapper;
import com.sprucetec.live.mcq.McqBusinessMemoryMap;
import com.sprucetec.live.service.*;
import com.sprucetec.live.utils.RedisUtil;
import com.sprucetec.mcq.producer.impl.MessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import tk.mybatis.mapper.entity.Example;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author chenkai
 * @date 2020/3/31 4:16 下午
 * @description:
 */
@Slf4j
@Service("liveVideoService")
public class LiveVideoServiceImpl implements LiveVideoService {
    private static final String limitMessage = "平台限流,请稍后再试";

    @Autowired
    private LiveVideoMapper liveVideoMapper;

    @Autowired
    private GenerateUniqueIdService generateUniqueIdService;

    @Autowired
    private LiveVideoRelateDetailService liveVideoRelateDetailService;

    @Autowired
    private LiveVideoHlsDetailService liveVideoHlsDetailService;

    @Autowired
    private LiveVideoServerDetailService liveVideoServerDetailService;

    @Autowired
    private MessageProducer mcqMessageProduct;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private LiveCategoryService liveCategoryService;

    @Autowired
    private LiveAnchorService liveAnchorService;

    @Qualifier("supportThreadPool")
    @Autowired
    ThreadPoolTaskExecutor supportThreadPool;

    @Autowired
    private LiveVideoDataService liveVideoDataService;

    @Autowired
    private LiveOperateLogService liveOperateLogService;


    @Autowired
    private LiveIntranetIpService liveIntranetIpService;

    @Value("${netty.server}")
    private String nettyServer;

    @Transactional(rollbackFor = SprucetecException.class)
    @LiveRedisCache(description = "更新redis相关数据")
    @Override
    public LiveVideo createLiveVideo(LiveVideoDto liveVideoDto) {
        this.checkVideoDetail(liveVideoDto);
        LiveVideo liveVideo = new LiveVideo();
        BeanUtils.copyProperties(liveVideoDto, liveVideo);
        String liveNo = generateUniqueIdService.getUniqueId(SerialNoPrefixEnum.LIVE);
        liveVideo.setLiveNo(liveNo);
        liveVideo.setCreateTime(new Date());
        if (VideoTypeEnum.ADVANCE.getCode() == liveVideo.getType()) { // 视频如果是预告，那么要设置一下视频状态
            liveVideo.setLiveStatus(LiveStatusEnum.NOLIVE.getCode());
        } else {
            liveVideo.setLiveStatus(LiveStatusEnum.LIVEEND.getCode());
        }
        liveVideo.setCreateUserCode(LoginUserContext.getUser().getUid());
        liveVideo.setCreateUserName(LoginUserContext.getUser().getName());
        liveVideoMapper.insertSelective(liveVideo);

        //插入子表
        for (LiveVideoServerDetail liveVideoServerDetail : liveVideoDto.getVideoServerDetailList()) {
            liveVideoServerDetail.setLiveVideoId(liveVideo.getId());
            liveVideoServerDetail.setCreateTime(new Date());
            liveVideoServerDetail.setCreateUserCode(LoginUserContext.getUser().getUid());
            liveVideoServerDetail.setCreateUserName(LoginUserContext.getUser().getName());
        }
        liveVideoServerDetailService.batchInsert(liveVideoDto.getVideoServerDetailList());
        for (LiveVideoHlsDetail liveVideoHlsDetail : liveVideoDto.getVideoHlsDetailList()) {
            liveVideoHlsDetail.setLiveVideoId(liveVideo.getId());
            liveVideoHlsDetail.setCreateTime(new Date());
            liveVideoHlsDetail.setCreateUserCode(LoginUserContext.getUser().getUid());
            liveVideoHlsDetail.setCreateUserName(LoginUserContext.getUser().getName());
        }
        liveVideoHlsDetailService.batchInsert(liveVideoDto.getVideoHlsDetailList());
        for (LiveVideoRelateDetail liveVideoRelateDetail : liveVideoDto.getVideoRelateDetailList()) {
            liveVideoRelateDetail.setLiveVideoId(liveVideo.getId());
            liveVideoRelateDetail.setCreateTime(new Date());
            liveVideoRelateDetail.setCreateUserCode(LoginUserContext.getUser().getUid());
            liveVideoRelateDetail.setCreateUserName(LoginUserContext.getUser().getName());
        }
        liveVideoRelateDetailService.batchInsert(liveVideoDto.getVideoRelateDetailList());
        //插入操作纪录
        liveVideoDto.setLiveNo(liveNo);
        insertOperationLog(liveVideoDto, LiveOperationTypeEnum.CREATE_LIVE.getCode());
        return liveVideo;
    }

    @Transactional(rollbackFor = SprucetecException.class)
    @LiveRedisCache(description = "更新redis相关数据")
    @Override
    public LiveVideo updateLiveVideo(LiveVideoDto liveVideoDto) {
        if (liveVideoDto.getId() == null) {
            throw new SprucetecException("id不能为空");
        }
        if (liveVideoDto.getLiveStatus().equals(LiveStatusEnum.LIVEING.getCode())) {
            throw new SprucetecException("正在直播的视频不能修改");
        }
        this.checkVideoDetail(liveVideoDto);
        LiveVideo liveVideo = new LiveVideo();
        BeanUtils.copyProperties(liveVideoDto, liveVideo);
        liveVideo.setUpdateTime(new Date());
        liveVideo.setUpdateUserCode(LoginUserContext.getUser().getUid());
        liveVideo.setUpdateUserName(LoginUserContext.getUser().getName());
        if (liveVideo.getType() == VideoTypeEnum.REVIEW.getCode()) { // 当设置回看的时候，也要把直播状态改成直播结束
            liveVideo.setLiveStatus(LiveStatusEnum.LIVEEND.getCode());
        }

        liveVideoMapper.updateByPrimaryKeySelective(liveVideo);

        for (LiveVideoServerDetail liveVideoServerDetail : liveVideoDto.getVideoServerDetailList()) {
            liveVideoServerDetail.setLiveVideoId(liveVideo.getId());
            liveVideoServerDetail.setCreateTime(new Date());
            liveVideoServerDetail.setCreateUserCode(LoginUserContext.getUser().getUid());
            liveVideoServerDetail.setCreateUserName(LoginUserContext.getUser().getName());
        }
        liveVideoServerDetailService.deleteServerDetail(liveVideoDto.getId());
        liveVideoHlsDetailService.deleteHlsDetail(liveVideoDto.getId());
        for (LiveVideoHlsDetail liveVideoHlsDetail : liveVideoDto.getVideoHlsDetailList()) {
            liveVideoHlsDetail.setLiveVideoId(liveVideo.getId());
            liveVideoHlsDetail.setCreateTime(new Date());
            liveVideoHlsDetail.setCreateUserCode(LoginUserContext.getUser().getUid());
            liveVideoHlsDetail.setCreateUserName(LoginUserContext.getUser().getName());
        }
        liveVideoServerDetailService.batchInsert(liveVideoDto.getVideoServerDetailList());
        liveVideoHlsDetailService.batchInsert(liveVideoDto.getVideoHlsDetailList());
        for (LiveVideoRelateDetail liveVideoRelateDetail : liveVideoDto.getVideoRelateDetailList()) {
            liveVideoRelateDetail.setLiveVideoId(liveVideo.getId());
            liveVideoRelateDetail.setCreateTime(new Date());
            liveVideoRelateDetail.setCreateUserCode(LoginUserContext.getUser().getUid());
            liveVideoRelateDetail.setCreateUserName(LoginUserContext.getUser().getName());
        }
        liveVideoRelateDetailService.deleteRelateDetail(liveVideoDto.getId());
        liveVideoRelateDetailService.batchInsert(liveVideoDto.getVideoRelateDetailList());
        //插入操作纪录
        insertOperationLog(liveVideoDto, LiveOperationTypeEnum.UPDATE_LIVE.getCode());
        return liveVideo;
    }

    @Override
    public LiveVideoDto getLiveVideoDto(Long id) {
        LiveVideo liveVideo = liveVideoMapper.selectByPrimaryKey(id);
        LiveVideoDto liveVideoDto = new LiveVideoDto();
        BeanUtils.copyProperties(liveVideo, liveVideoDto);

        Map<Long, LiveCategory> allCategoryMap = liveCategoryService.getAllCategoryMap();
        LiveCategory category = allCategoryMap.getOrDefault(liveVideo.getCategoryId(), new LiveCategory());
        liveVideoDto.setCategoryName(category.getCategoryName());

        Map<Long, LiveAnchor> allAnchorMap = liveAnchorService.getAllAnchorMap();
        LiveAnchor anchor = allAnchorMap.getOrDefault(liveVideo.getAnchorId(), new LiveAnchor());
        liveVideoDto.setAnchorName(anchor.getAnchorName());

        List<LiveVideoRelateDetail> relateDetailList = liveVideoRelateDetailService.getRelateDetailList(liveVideo.getId());
        liveVideoDto.setVideoRelateDetailList(relateDetailList);

        List<LiveVideoServerDetail> serverDetailList = liveVideoServerDetailService.getServerDetailList(liveVideo.getId());
        liveVideoDto.setVideoServerDetailList(serverDetailList);

        List<LiveVideoHlsDetail> hlsDetailList = liveVideoHlsDetailService.getHlsDetailList(liveVideo.getId());
        liveVideoDto.setVideoHlsDetailList(hlsDetailList);
        return liveVideoDto;
    }

    /**
     * 校验视频明细
     *
     * @param liveVideoDto
     */
    private void checkVideoDetail(LiveVideoDto liveVideoDto) {

        if (CollectionUtils.isEmpty(liveVideoDto.getVideoServerDetailList())) {
            throw new SprucetecException("视频服务器不能为空");
        }
        if (CollectionUtils.isEmpty(liveVideoDto.getVideoHlsDetailList())) {
            throw new SprucetecException("视频Hls不能为空");
        }
        if (CollectionUtils.isEmpty(liveVideoDto.getVideoServerDetailList())) {
            throw new SprucetecException("关联视频不能为空");
        }
    }

    /**
     * 插入操作纪录
     *
     * @param liveVideoDto
     */
    private void insertOperationLog(LiveVideoDto liveVideoDto, Integer operationType) {
        StringBuilder sb = new StringBuilder();
        sb.append("操作人:" + "=>" + LoginUserContext.getUser().getUid() + "-" + LoginUserContext.getUser().getName());
        LiveOperateLog log = new LiveOperateLog();
        log.setBillNo(liveVideoDto.getLiveNo());
        log.setModule(LiveOperationTypeEnum.code2Desc(operationType));
        log.setOperateType(operationType);
        log.setOperateTime(new Date());
        log.setContent(LiveOperationTypeEnum.code2Desc(operationType) + ": " + sb);
        log.setCreateTime(new Date());
        log.setCreateUserCode(LoginUserContext.getUser().getUid());
        log.setCreateUserName(LoginUserContext.getUser().getName());
        liveOperateLogService.insertOperateLog(log);
    }

    @Override
    public PagedListDTO<LiveVideoDto> getLiveVidePage(LiveVideoDto liveVideoDto) {
        PagedListDTO<LiveVideoDto> pagedListDTO = new PagedListDTO<>();
        PageHelper.startPage(liveVideoDto.getPageIndex(), liveVideoDto.getPageSize());
        Example example = new Example(LiveVideo.class);
        example.setOrderByClause("create_time desc");
        Example.Criteria criteria = example.createCriteria();
        criteria.andEqualTo("isDeleted", 0);
        if (liveVideoDto.getAnchorId() != null) {
            criteria.andLike("anchorId", String.valueOf(liveVideoDto.getAnchorId()));
        }
        if (StringUtils.isNotBlank(liveVideoDto.getTitle())) {
            criteria.andLike("title", "%" + liveVideoDto.getTitle() + "%");

        }
        if (liveVideoDto.getTypeList().size() != 0) {
            criteria.andIn("type", liveVideoDto.getTypeList());
        }
        if (liveVideoDto.getCategoryIdList().size() != 0) {
            criteria.andIn("categoryId", liveVideoDto.getCategoryIdList());
        }
        Map<Long, LiveCategory> allCategoryMap = liveCategoryService.getAllCategoryMap();
        Map<Long, LiveAnchor> allAnchorMap = liveAnchorService.getAllAnchorMap();
        List<LiveVideo> liveVideoList = liveVideoMapper.selectByExample(example);
        List<LiveVideoDto> liveVideoDtoList = new ArrayList<>(liveVideoList.size());
        for (LiveVideo liveVideo : liveVideoList) {
            LiveVideoDto videoDto = new LiveVideoDto();
            BeanUtils.copyProperties(liveVideo, videoDto);
            LiveCategory category = allCategoryMap.getOrDefault(liveVideo.getCategoryId(), new LiveCategory());
            videoDto.setCategoryName(category.getCategoryName());
            LiveAnchor anchor = allAnchorMap.getOrDefault(liveVideo.getAnchorId(), new LiveAnchor());
            videoDto.setAnchorName(anchor.getAnchorName());
            liveVideoDtoList.add(videoDto);
        }
        PageInfo<LiveVideo> pages = new PageInfo(liveVideoDtoList);
        pagedListDTO.setTotal((int) pages.getTotal());
        pagedListDTO.setRecords(liveVideoDtoList);
        PageDTO pageDTO = new PageDTO();
        pageDTO.setPageSize(liveVideoDto.getPageSize());
        pageDTO.setPageIndex(liveVideoDto.getPageIndex());
        pagedListDTO.setPage(pageDTO);
        return pagedListDTO;
    }

    @Value("${system.high.concurrency.category}")
    private String isHighConcurrncyCategory; // MCQ聊天消息消费模式

    @LiveRedisCache(description = "更新redis相关数据")
    @Override
    public LiveVideoDto setVideo(LiveVideoDto liveVideoDto) throws Exception {
        LiveVideo liveVideo = new LiveVideo();
        BeanUtils.copyProperties(liveVideoDto, liveVideo);
        switch (liveVideoDto.getOperateType()) {
            case "carousel":
                Integer isCarousel = liveVideoDto.getVideoSetting().getIsCarousel();
                liveVideo.setIsCarousel(isCarousel);
                insertOperationLog(liveVideoDto, isCarousel == 0 ? LiveOperationTypeEnum.CANCEL_CAROUSEL.getCode() : LiveOperationTypeEnum.SET_CAROUSEL.getCode());
                break;
            case "top":
                Integer isTop = liveVideoDto.getVideoSetting().getIsTop();
                liveVideo.setIsTop(isTop);
                insertOperationLog(liveVideoDto, isTop == 0 ? LiveOperationTypeEnum.CANCEL_TOP.getCode() : LiveOperationTypeEnum.TOP_LIVE.getCode());
                break;
            case "status":
                supportThreadPool.execute(() -> {
                    try {
                        this.toNettyServer(liveVideoDto); // 通知各个客户端当前直播被关闭
                    } catch (Exception e) {
                        log.error("关闭直播的消息通知失败，失败原因=>", e);
                    }
                });
                Integer liveStatus = liveVideoDto.getVideoSetting().getLiveStatus();
                liveVideo.setLiveStatus(liveStatus);
                insertOperationLog(liveVideoDto, liveStatus == 1 ? LiveOperationTypeEnum.UP_SHELVES.getCode() : LiveOperationTypeEnum.CLOSE_LIVE.getCode());
                if (liveVideoDto.getLiveStatus().equals(LiveStatusEnum.LIVEING.getCode())) { // 关闭按钮触发
                    closeVideo(liveVideoDto, liveVideo);
                } else if (liveVideoDto.getLiveStatus().equals(LiveStatusEnum.LIVEEND.getCode())) { // 重新开启直播
                    reOpenVideo(liveVideoDto, liveVideo);
                }
                break;
            case "delete":
                liveVideo.setIsDeleted(liveVideoDto.getVideoSetting().getIsDeleted());
                liveVideo.setIsShelves(0); // 删除也会被自动下架
                insertOperationLog(liveVideoDto, LiveOperationTypeEnum.DELETE_LIVE.getCode());
                break;
            case "shelves":
                Integer isShelves = liveVideoDto.getVideoSetting().getIsShelves();
                liveVideo.setIsShelves(isShelves);
                insertOperationLog(liveVideoDto, isShelves == 0 ? LiveOperationTypeEnum.DOWN_SHELVES.getCode() : LiveOperationTypeEnum.UP_SHELVES.getCode());
        }
        liveVideoMapper.updateByPrimaryKey(liveVideo);
        return liveVideoDto;
    }

    @Autowired
    private RedisUtil redisUtil;


    /**
     * 关闭直播生命周期
     */
    private void closeVideo(LiveVideoDto liveVideoDto, LiveVideo liveVideo) {
        liveVideo.setType(VideoTypeEnum.REVIEW.getCode());
        this.statisticalVideoData(liveVideoDto); // 统计直播数据

        if (isHighConcurrncyCategory.equals(LiveHighConcurrncyCategoryEnum.SCHEDULED.getType())) {
            ScheduledExecutorService scheduledExecutorService = McqBusinessMemoryMap.scheduledExecutorServiceMap.get(liveVideo.getId());
            if (scheduledExecutorService != null) {
                scheduledExecutorService.shutdown();
                McqBusinessMemoryMap.scheduledExecutorServiceMap.remove(liveVideo.getId());
                log.info("成功关闭定时任务;编号=>{}", liveVideoDto.getLiveNo());
            }
        }
    }

    /**
     * 重新开启直播生命周期
     *
     * @param liveVideoDto
     * @param liveVideo
     */
    private void reOpenVideo(LiveVideoDto liveVideoDto, LiveVideo liveVideo) {
        liveVideo.setType(VideoTypeEnum.ADVANCE.getCode());
        log.info("重置redis热点数据超时时间");
        this.coverRedisExpire(liveVideoDto, liveVideo); // 重新设置redis数据生命周期超时时间
    }

    private void coverRedisExpire(LiveVideoDto liveVideoDto, LiveVideo liveVideo) {
        Integer noExpireTime = Integer.MAX_VALUE;
        String chatKey = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveVideoDto.getId() + ":type:0";
        String questionKey = RedisKeyConstant.LiveChat.LIVE_CHAT_CACHE + ":videoId:" + liveVideoDto.getId() + ":type:1";
        String watchNumKey = RedisKeyConstant.LiveVideoDataStatistics.watchNum + liveVideoDto.getLiveNo();
        String onlineUserKey = RedisKeyConstant.LiveVideoDataStatistics.onlineUser + liveVideoDto.getLiveNo();
        String topNumKey = RedisKeyConstant.LiveVideoDataStatistics.topNum + liveVideoDto.getLiveNo();
        redisUtil.expire(chatKey, noExpireTime, TimeUnit.SECONDS);
        redisUtil.expire(questionKey, noExpireTime, TimeUnit.SECONDS);
        redisUtil.expire(topNumKey, noExpireTime, TimeUnit.SECONDS);
        redisUtil.expire(onlineUserKey, noExpireTime, TimeUnit.SECONDS);
        redisUtil.expire(watchNumKey, noExpireTime, TimeUnit.SECONDS);
    }

    /**
     * 统计直播数据
     *
     * @param liveVideoDto 直播实体类
     */
    private void statisticalVideoData(LiveVideoDto liveVideoDto) {
        supportThreadPool.execute(() -> {
            liveVideoDataService.statisticalVideoData(liveVideoDto);
        });
    }

    /**
     * 通过MQ发送消息关闭直播
     *
     * @param liveVideoDto
     * @throws Exception
     */
    // 根据实际websocket地址更改
    private void toNettyServer(LiveVideoDto liveVideoDto) throws Exception {
        // 下面是单节点的发送数据到netty服务器
        /*WsClient myClient = new WsClient(new URI(nettyServer));
        myClient.connect();
        // 判断是否连接成功，未成功后面发送消息时会报错
        while (!myClient.getReadyState().equals(WebSocket.READYSTATE.OPEN)) {
            System.out.println("连接中···请稍后");
            Thread.sleep(1000);
        }
        JSONObject message = new JSONObject();
        if (liveVideoDto.getLiveStatus() == 0 || liveVideoDto.getLiveStatus() == 2) { // 开启直播
            message.put("content", "open");
        } else { // 关闭直播
            message.put("content", "close");
        }
        message.put("videoId", liveVideoDto.getId());
        message.put("type", "LIVE_SERVER");
        myClient.send(message.toJSONString());*/
        // 发送消息到MQ， 通过MQ关闭直播
        NettyMessageDto nettyMessageDto = new NettyMessageDto();
        nettyMessageDto.setVideoId(liveVideoDto.getId());
        if (liveVideoDto.getLiveStatus() == LiveStatusEnum.NOLIVE.getCode() || liveVideoDto.getLiveStatus() == LiveStatusEnum.LIVEEND.getCode()) { // 开启直播
            nettyMessageDto.setContent("open");
        } else { // 关闭直播
            nettyMessageDto.setContent("close");
        }
        Message message = new Message(MqConstant.TOPIC, MqConstant.NettyMessageMq.TAG_NETTY_VIDEO_CLOSE,
                MqConstant.NettyMessageMq.TAG_NETTY_VIDEO_CLOSE_KEY + new Date().getTime(),
                JSON.toJSONString(nettyMessageDto).getBytes(StandardCharsets.UTF_8));
        mcqMessageProduct.send(message);
    }

    @Autowired
    private LiveOnlineUserService liveOnlineUserService;

    /**
     * 导出在线用户数
     *
     * @param liveVideo
     * @return
     */
    @Override
    public List<LiveUserDto> getOnlineUserByLiveNo(LiveVideo liveVideo) {
        // 判断当前直播是否已经关闭，如果没有关闭则从redis中取在线用户数
        Example example = new Example(LiveVideo.class);
        example.createCriteria().andEqualTo("liveNo", liveVideo.getLiveNo());
        LiveVideo lv = liveVideoMapper.selectByExample(example).get(0);
        if (lv.getLiveStatus() == LiveStatusEnum.LIVEING.getCode()) {//正在直播
            return redisTemplate.boundHashOps(RedisKeyConstant.LiveVideoDataStatistics.onlineUser + liveVideo.getLiveNo())
                    .values();
        } else if (lv.getLiveStatus() == LiveStatusEnum.LIVEEND.getCode()) { // 直播结束,从mysql中取在线人数
            List<LiveOnlineUser> list = liveOnlineUserService.getOnlineUserByLiveNo(liveVideo.getLiveNo());
            List<LiveUserDto> liveUserDtos = new ArrayList<>();
            list.stream().forEach((liveOnlineUser) -> {
                LiveUserDto liveUserDto = new LiveUserDto();
                BeanUtils.copyProperties(liveOnlineUser, liveUserDto);
                liveUserDto.setLiveNo(liveOnlineUser.getRelateVideoNo());
                liveUserDtos.add(liveUserDto);
            });
            return liveUserDtos;
        } else {
            throw new SprucetecException("直播状态问题，导出失败");
        }
    }

    /************************************************************ PC 端相关接口 START ************************************************************/
    private static final String LOCK_KEY = "LIVE:LIVE_LOCK_RESOURCE";

    @Autowired
    private DistributedLocker distributedLocker;

    /**
     * 查询文章详情
     * 互斥锁 队列防止缓存雪崩 但是会降低吞吐量 不是特别高的并发可以使用
     *
     * @param liveNo
     * @return
     */
    @Override
    public LiveVideoDto getVideoDetail(String liveNo, String ip, String city) {
        LiveVideoDto liveVideoDto = null;
        Object o = redisTemplate.boundHashOps(RedisKeyConstant.LiveVideoResources.VIDEO_DETAIL).get(liveNo);
        if (o == null) {
            String lockKey = LOCK_KEY + ":videoNo=" + liveNo;
            try {
                boolean locked = distributedLocker.tryLock(lockKey, TimeUnit.SECONDS, 0L, 30);
                if (locked) {
                    LiveVideoDto vd = new LiveVideoDto();
                    vd.setLiveNo(liveNo);
                    liveVideoDto = liveVideoMapper.getVideoDetail(vd);
                    String liveHlsAddress = liveVideoDto.getLiveHlsAddress();
                    String advances = liveVideoDto.getIsAdvance();
                    String intranets = liveVideoDto.getIsIntranet();
                    if (liveHlsAddress == null) throw new SprucetecException("该视频没有配置HLS视频流地址，请联系管理员");
                    String[] liveHlsArrressArr = liveHlsAddress.split(",");
                    String[] liveHlsType = liveVideoDto.getLiveHlsType().split(",");
                    String[] advanceArr = advances.split(",");
                    String[] intranet = intranets.split(",");
                    List<LiveVideoHlsDetail> liveVideoHlsDetails = new ArrayList<>();
                    for (int i = 0; i < liveHlsArrressArr.length; i++) {
                        LiveVideoHlsDetail liveVideoHlsDetail = new LiveVideoHlsDetail();
                        liveVideoHlsDetail.setLiveHlsType(liveHlsType[i]);
                        liveVideoHlsDetail.setLiveHlsAddress(liveHlsArrressArr[i]);
                        liveVideoHlsDetail.setIsAdvance(Integer.parseInt(advanceArr[i]));
                        liveVideoHlsDetail.setIsIntranet(Integer.parseInt(intranet[i]));
                        liveVideoHlsDetails.add(liveVideoHlsDetail);
                    }
                    liveVideoDto.setVideoHlsDetailList(liveVideoHlsDetails);
                    redisTemplate.boundHashOps(RedisKeyConstant.LiveVideoResources.VIDEO_DETAIL).put(liveNo, liveVideoDto);
                    return ipRuleMapping(liveVideoDto, ip);
                } else {
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return getVideoDetail(liveNo, ip, city);
                }
            } finally {
                distributedLocker.unlock(lockKey);
            }
        } else {
            return ipRuleMapping((LiveVideoDto) o, ip);
        }
    }


    /**
     * 通过IP 匹配对应的HLS的地址
     *
     * @param liveVideoDto
     * @Param ip 请求IP
     */
    private LiveVideoDto ipRuleMapping(LiveVideoDto liveVideoDto, String ip) {
        List<LiveVideoHlsDetail> rellyList = new ArrayList<>();
        if (JvmCache.ipList.size() == 0) { // 如果jvm缓存不存在， 则从redis中查询
            Set members = redisTemplate.boundSetOps(RedisKeyConstant.LiveVideoResources.INTRANET_IP).members();
            if (!CollectionUtils.isEmpty(members)) {
                members.forEach(x -> {
                    ((JSONArray) x).forEach(m -> {
                        JvmCache.ipList.add(m.toString());
                    });
                });
            } else {
                List<LiveIntranetIp> liveIntranetIps = liveIntranetIpService.getLiveList();
                List<String> list = new ArrayList<>();
                liveIntranetIps.forEach(x -> {
                    JvmCache.ipList.add(x.getIp());
                    list.add(x.getIp());
                });
                redisTemplate.boundSetOps(RedisKeyConstant.LiveVideoResources.INTRANET_IP).add(list);
            }
        }
        if (JvmCache.ipList.contains(ip)) { //是内网IP
            List<LiveVideoHlsDetail> videoHlsDetailList = liveVideoDto.getVideoHlsDetailList();
            videoHlsDetailList.forEach(x -> {
                if (LiveIntranetEnum.YES.getCode().equals(x.getIsIntranet())) {
                    rellyList.add(x);
                }
            });
            log.info("内网用户登录， ip: {}", ip);
        } else {
            List<LiveVideoHlsDetail> videoHlsDetailList = liveVideoDto.getVideoHlsDetailList();
            videoHlsDetailList.forEach(x -> {
                if (LiveIntranetEnum.NO.getCode().equals(x.getIsIntranet())) {
                    rellyList.add(x);
                }
            });
            log.info("外网用户登录， ip: {}", ip);
        }
        liveVideoDto.setVideoHlsDetailList(rellyList);
        return liveVideoDto;
    }

    /**
     * 通过城市匹配
     *
     * @param liveVideoDto
     */
    private static LiveVideoDto cityRuleMapping(LiveVideoDto liveVideoDto, String city, String ip) {
        List<LiveVideoHlsDetail> rellyList = new ArrayList<>();

        if (!SprucetecConstant.consumerCity.contains(city)) { //是内网IP
            List<LiveVideoHlsDetail> videoHlsDetailList = liveVideoDto.getVideoHlsDetailList();
            videoHlsDetailList.forEach(x -> {
                if (LiveIntranetEnum.YES.getCode().equals(x.getIsIntranet())) {
                    rellyList.add(x);
                }
            });
            log.info("内网用户登录，用户所在地{}, ip: {}", city, ip);
        } else {
            List<LiveVideoHlsDetail> videoHlsDetailList = liveVideoDto.getVideoHlsDetailList();
            videoHlsDetailList.forEach(x -> {
                if (LiveIntranetEnum.NO.getCode().equals(x.getIsIntranet())) {
                    rellyList.add(x);
                }
            });
            log.info("外网用户登录，用户所在地{}, ip: {}", city, ip);
        }
        liveVideoDto.setVideoHlsDetailList(rellyList);
        return liveVideoDto;
    }


    @Override
    public void videoDetailData(LiveVideoDto liveVideoDto) {
        Long onlineUserNum = redisTemplate.boundValueOps(MqConstant.ChatMq.onlineUserNum + liveVideoDto.getVideoId()).increment(1);// 在线人数加 1
        liveVideoDto.setOnlineUserNum(onlineUserNum.toString());
        // 异步 添加查询视频详情，更新相关数据
        Message message = new Message(MqConstant.TOPIC, MqConstant.VideoDataMq.TAG,
                MqConstant.VideoDataMq.KEY + new Date().getTime(), JSON.toJSONString(liveVideoDto).getBytes(StandardCharsets.UTF_8));
        try {
            mcqMessageProduct.send(message);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Integer getRemainsCountByVideo(LiveChat liveChat) {
        Example example = new Example(LiveVideo.class);
        example.createCriteria().andEqualTo("id", liveChat.getVideoId())
                .andEqualTo("questionLimit", 1);// 并且是提问限制的，如果不限制则不需要控制
        List<LiveVideo> liveVideos = liveVideoMapper.selectByExample(example);
        if (!CollectionUtils.isEmpty(liveVideos)) {
            return liveVideos.get(0).getQuestionCount();
        }
        return Integer.MAX_VALUE;
    }

    @Value("${redis.queue}")
    private boolean redisQueue;

    @Autowired
    private RedisQueueService redisQueueService;

    @Override
    public Integer clearChannel(NettyMessageDto nettyMessageDto) {
        try {
            if (!redisQueue) { //MCQ 消息队列方案
                Message message = new Message(MqConstant.TOPIC, MqConstant.NettyMessageMq.TAG_NETTY_CHANNEL_REMOVE,
                        MqConstant.NettyMessageMq.TAG_NETTY_CHANNEL_REMOVE_KEY + new Date().getTime(),
                        JSON.toJSONString(nettyMessageDto).getBytes(StandardCharsets.UTF_8));
                mcqMessageProduct.send(message);
            } else { //  redis-queue 消息队列方案
                redisQueueService.sendChatClose(JSONObject.toJSONString(nettyMessageDto));
            }
        } catch (Exception e) {
            e.printStackTrace();
            return 0; // 发送失败
        }
        return 1;
    }

    /**
     * 加载PC端首页数据
     * todo 这里会有会有缓存击穿问题 因为时间有限先做一下限流。后面会用 guava 布隆过滤器防止 缓存击穿
     *
     * @param liveUser
     * @return
     */
    private Semaphore indexPageSemaphore = new Semaphore(50);// 允许执行指定代码块的线程数量
    private static final Long limitTime = 2L;

    @Override
    public Map getIndexVideo(LiveUser liveUser) {
        Map resultMap = null;
        try {
            if (indexPageSemaphore.tryAcquire(limitTime, TimeUnit.SECONDS)) { // 等待2秒钟 不能得到执行权，触发限流
                resultMap = new HashMap();
                // 获取置顶数据
                List<LiveVideoDto> topLiveVideo = redisTemplate.boundHashOps(RedisKeyConstant.LiveKind.TOPVIDEO).values();
                if (CollectionUtils.isEmpty(topLiveVideo)) {
                    LiveVideoDto liveVideoDto = new LiveVideoDto();
                    liveVideoDto.setIsTop(1);
                    liveVideoDto.setIsShelves(1);
                    liveVideoDto.setIsDeleted(LiveSettingEnum.NOT_DELETE.getCode());
                    topLiveVideo = liveVideoMapper.getIndexVideo(liveVideoDto);
                    Map<Long, LiveVideoDto> collect = topLiveVideo.stream().collect(Collectors.toMap(LiveVideo::getId, Function.identity()));
                    redisTemplate.boundHashOps(RedisKeyConstant.LiveKind.TOPVIDEO).putAll(collect);
                }
                // 获取最新视频
                List<LiveVideoDto> newLastedVideos = getNewLastedVideos(false);
                // 获取轮播视频
                List<LiveVideoDto> carouselVideos = redisTemplate.boundHashOps(RedisKeyConstant.LiveKind.CAROUSEL).values();
                if (CollectionUtils.isEmpty(carouselVideos)) {
                    LiveVideoDto liveVideoDto = new LiveVideoDto();
                    //liveVideoDto.setIsTop(0);
                    liveVideoDto.setIsShelves(LiveSettingEnum.UP_SHELVES.getCode());
                    liveVideoDto.setIsCarousel(LiveSettingEnum.CAROUSEL.getCode());
                    liveVideoDto.setIsDeleted(LiveSettingEnum.NOT_DELETE.getCode());
                    carouselVideos = liveVideoMapper.getIndexVideo(liveVideoDto);
                    Map<Long, LiveVideoDto> collect = carouselVideos.stream().collect(Collectors.toMap(LiveVideo::getId, Function.identity()));
                    redisTemplate.boundHashOps(RedisKeyConstant.LiveKind.CAROUSEL).putAll(collect);
                }
                resultMap = buildIndexData(resultMap, topLiveVideo, newLastedVideos, carouselVideos);
            } else {
                throw new SprucetecException(limitMessage);
            }
        } catch (Exception ex) {
            throw new SprucetecException(ex.toString());
        } finally {
            indexPageSemaphore.release();
        }
        return resultMap;
    }

    private Map buildIndexData(Map resultMap, List<LiveVideoDto> topLists, List<LiveVideoDto> newLastedVideos, List<LiveVideoDto> carouselVideos) {
        if (topLists.size() > 0) {
            resultMap.put("mainVideo", topLists.get(0));
            topLists.remove(0);
            resultMap.put("subVideo", topLists);
        } else {
            resultMap.put("mainVideo", null);
            resultMap.put("subVideo", new ArrayList<>());
        }
        resultMap.put("newLasted", newLastedVideos);
        resultMap.put("carousel", carouselVideos);
        return resultMap;
    }

    /**
     * PC端搜索视频 // todo 这里最好做一下redis限流，后面有时间再优化
     *
     * @param liveVideoDto
     * @return
     */
    private Semaphore semaphore = new Semaphore(10); // 只能同时查10 个并发，其他的等待，减少恶意请求对数据库的压力破坏

    @Override
    public List<LiveVideoDto> videoSearch(LiveVideoDto liveVideoDto) throws InterruptedException {

        List<LiveVideoDto> liveVideos = new ArrayList<>();
        try {
            if (semaphore.tryAcquire(limitTime, TimeUnit.SECONDS)) {
                liveVideoDto.setIsShelves(LiveSettingEnum.UP_SHELVES.getCode());
                liveVideoDto.setIsDeleted(LiveSettingEnum.NOT_DELETE.getCode());
                liveVideos = liveVideoMapper.getIndexVideo(liveVideoDto);
            } else {
                throw new SprucetecException(limitMessage);
            }
        } catch (Exception ex) {
            throw new SprucetecException(ex.toString());
        } finally {
            semaphore.release();
        }
        return liveVideos;
    }

    @Override
    public List<LiveVideoDto> getNewLastedVideos(boolean isDetail) {
        List<LiveVideoDto> newLastedVideos = redisTemplate.boundHashOps(RedisKeyConstant.LiveKind.NEWLAST).values();
        if (CollectionUtils.isEmpty(newLastedVideos)) {
            String lockKey = LOCK_KEY + ":newLasted";
            try {
                boolean locked = distributedLocker.tryLock(lockKey, TimeUnit.SECONDS, 0L, 30);
                if (locked) {
                    LiveVideoDto liveVideoDto = new LiveVideoDto();
                    liveVideoDto.setIsShelves(LiveSettingEnum.UP_SHELVES.getCode());
                    liveVideoDto.setIsDeleted(LiveSettingEnum.NOT_DELETE.getCode());
                    newLastedVideos = liveVideoMapper.getIndexVideo(liveVideoDto);
                    Map<Long, LiveVideoDto> collect = newLastedVideos.stream().collect(Collectors.toMap(LiveVideo::getId, Function.identity()));
                    redisTemplate.boundHashOps(RedisKeyConstant.LiveKind.NEWLAST).putAll(collect);
                    return limitFiveNews(isDetail, newLastedVideos);
                } else {
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return getNewLastedVideos(isDetail);
                }
            } finally {
                distributedLocker.unlock(lockKey);
            }
        } else {
            return limitFiveNews(isDetail, newLastedVideos);
        }
    }

    private List<LiveVideoDto> limitFiveNews(boolean isDetail, List<LiveVideoDto> newLastedVideos) {
        if (isDetail) {// 详情页默认只展示5条最新视频
            if (newLastedVideos.size() >= 5) {
                return newLastedVideos.subList(newLastedVideos.size() - 6, newLastedVideos.size() - 1);
            } else {
                return newLastedVideos;
            }
        } else {
            return newLastedVideos;
        }
    }

    /************************************************************ PC 端相关接口 END ************************************************************/

}
